package com.xbai.spark.recall.data

import com.xbai.spark.recall.util.SparkSessionUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * @author xbai
  * @Date 2021/1/11
  */
object FileDataLoader {

  val userPath = "/data/user_profile.txt"
  val itemPath = "/data/music_meta.txt"
  val dataPath = "/data/user_watch_pref.txt"

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSessionUtil.getSession("recommend", "recommend")
    getUserData(spark)
    getItemData(spark)
    getUserRemainData(spark)
  }

  def getUserData(sparkSession: SparkSession): DataFrame = {
    val textRDD: RDD[String] = sparkSession.sparkContext.textFile(userPath)
    println(textRDD.take(3).mkString)
    val userRDD: RDD[User] = textRDD.map(_.split("\t")).filter(_.length >= 5)
      .map(attrs => User(attrs(0), attrs(1), attrs(2), attrs(3), attrs(4)))
    val userDF: DataFrame = sparkSession.createDataFrame(userRDD)
    userDF.show(3, truncate = false)
    userDF.write.mode("overwrite").saveAsTable("muser")
    userDF
  }

  def getItemData(sparkSession: SparkSession): DataFrame = {
    val textRDD: RDD[String] = sparkSession.sparkContext.textFile(itemPath)
    val itemRDD: RDD[Item] = textRDD.map(_.split("\t")).filter(_.length >= 5)
      .map(attrs => Item(attrs(0), attrs(1), attrs(2), attrs(3), attrs(4)))
    val itemDF: DataFrame = sparkSession.createDataFrame(itemRDD)
    itemDF.show(3, truncate = false)
    itemDF.write.mode("overwrite").saveAsTable("mitem")
    itemDF
  }

  def getUserRemainData(sparkSession: SparkSession): DataFrame = {
    val textRDD: RDD[String] = sparkSession.sparkContext.textFile(dataPath)
    val accessRDD: RDD[AssessData] = textRDD.map(_.split("\t")).filter(_.length >= 4)
      .map(attrs => AssessData(attrs(0), attrs(1), attrs(2), attrs(3)))
    val accessDF: DataFrame = sparkSession.createDataFrame(accessRDD)
    accessDF.show(3, truncate = false)
    accessDF.write.mode("overwrite").saveAsTable("mremain")
    accessDF
  }

  /**
    * user_profile 初始化可以不用new
    *
    * @param userId   用户id
    * @param gender   性别
    * @param age      年龄段
    * @param salary   停留时间区间
    * @param location 地区省份
    */
  case class User(userId: String, gender: String, age: String, salary: String, location: String)

  /**
    * music_meta
    *
    * @param musicId  音乐id
    * @param desc     简单描述信息
    * @param category 类别
    * @param location 地区
    * @param labels   音乐标签
    */
  case class Item(musicId: String, desc: String, category: String, location: String, labels: String)

  /**
    * user_watch_pref
    *
    * @param userId       用户id
    * @param musicId      音乐id
    * @param remainTime   收听时长（单位秒）
    * @param durationHour 收听时间点（hour）
    */
  case class AssessData(userId: String, musicId: String, remainTime: String, durationHour: String)

}
